package com.demo.rocketmq.producer.ha;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQUtil;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
@Service
public class MasterSlave {
    @Resource(name = "rocketMQTemplate")
    private RocketMQTemplate rocketMqTemplate;

    @Resource
    private RocketMQProperties rocketMQProperties;

    public void send() {
        try {
            for (int i = 0; i < 10000; i++) {
                rocketMqTemplate.syncSend("test-topic-1", String.format("第%d条消息,日期：%s", i, new Date()));
            }
        } catch (Exception e) {
            log.warn("broker异常");
        }
        log.info("发送消息完毕");
    }

    /**
     * 测试消费master/slave broker消息<br>
     * 场景一：<br>
     * 消费master broker消息时，master broker宕机，预期从slave broker消费<br>
     * 场景二：<br>
     * 接着场景一，继续从slave消费，预期没有消息可以消费<br>
     * 场景三：<br>
     * 接着场景一，消费者从slave broker消费完消息后，消费者消费重启，master broker也从宕机中恢复正常，预期消费者消费master broker宕机前未消费的消息<br>
     * 场景四：<br>
     * 接着场景一，消费者从slave broker消费消息过程中，master broker从宕机中恢复，预期结果继续消费未消费的消息，偏移量以slave broker为准，但消息从master borker拉取
     *
     * @throws MQClientException 异常
     * @throws InterruptedException 异常
     */
    public void read() throws MQClientException, InterruptedException {
        log.info("开始消费");
        Map<String, AtomicInteger> map = new HashMap<>();
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("read_slave", true, StringUtils.EMPTY);
        consumer.setInstanceName(RocketMQUtil.getInstanceName(rocketMQProperties.getNameServer()));
        consumer.setNamesrvAddr(rocketMQProperties.getNameServer());
        consumer.setConsumeThreadMax(64);
        consumer.setMessageModel(MessageModel.CLUSTERING);
        consumer.subscribe("test-topic-1", "*");
        consumer.setMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            msgs.forEach(msg -> {
                try {
                    TimeUnit.MILLISECONDS.sleep(50);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }

                String text = new String(msg.getBody());
                AtomicInteger atomicInteger = map.getOrDefault(text, new AtomicInteger());
                atomicInteger.incrementAndGet();
                map.put(text, atomicInteger);
                log.info("read_slave消费者消费的消息：{}", text);
            });
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        consumer.start();
        TimeUnit.MINUTES.sleep(6);
        map.forEach((key, value) -> {
            if (value.get() > 1) {
                log.info("重复消费消费：{}，消费次数：{}", key, value.get());
            }
        });
        log.info("消费完毕，消费消息数量：{}", map.size());
    }
}
